package com.runa.monitor.platinterface.mqtt;

import com.alibaba.fastjson.JSON;
import com.runa.monitor.comm.dto.newComm.AsynResult;
import com.runa.monitor.platinterface.service.GwService;
import com.runa.persistence.DateUtil;
import com.runa.persistence.entity.gw.GwDown;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;

/**
 * 2024/7/11 10:56
 * zhangYB
 */

@Slf4j
@Component
@EnableScheduling
public class MqttSchedule {

    @Resource(name = "asyncRequestScheduler")
    private Executor executor;

    private static ConcurrentLinkedQueue<MqttSendMessage> queue = new ConcurrentLinkedQueue<>();

    private static volatile Map<String, AtomicLong> topicMap = new ConcurrentHashMap<>();

    private static final int MAX_QUEUE_SIZE = 100 * 10000;
    private static AtomicInteger size = new AtomicInteger(0);


    @Resource
    private GwService gwService;

    public MqttSchedule() {
    }

    public MqttSchedule(Executor executor) {
        this.executor = executor;
    }

    /**
     * 网关校时
     * {"Func":"down","Id":"24001131","Param":{"Data":{"Cmd":"setdatetime","Value":"2025-1-14 09:21:27"},"Type":"ctrl"},"Seq":"9365982476","Time":115114105}
     */
    @Scheduled(cron="${aigw.timing.interval:0 0 0/1 * * ?}")
    public void timing() {
        gwService.downTiming("FFFFFFF");
    }

    @Scheduled(cron = "0 0/1 * * * ?")
    public void execute() {
        send();
//        printTopicStatistics();
    }


    public void send() {
        try {
            int queueSize = queue.size();
            log.debug("定时处理缓存MQTT消息: {} {}", queueSize, size.get());
            int n = 0;
            while (n < EmqConfig.getScheduleQueueCount() || n < queueSize) {
                MqttSendMessage sendMessage = queue.poll();
                if (null != sendMessage) {
                    size.decrementAndGet();
                    String topic = sendMessage.getTopic();
                    MqttMessage message = sendMessage.getMessage();
                    executor.execute(() -> MqttBaseController.getConnector().sendMessage(topic, message));
                    n++;
                } else {
                    break;
                }
            }
            log.debug("定时处理缓存MQTT消息结束, 本次处理: {}", n);
        } catch (Exception e) {
            log.error("send: {}", e);
        }
    }

    public static void addToQueue(String topic, MqttMessage message) {
        if (size.get() > MAX_QUEUE_SIZE) {
            log.error("队列已满");
            return;
        }
        try {
            MqttSendMessage sendMessage = new MqttSendMessage();
            sendMessage.setTopic(topic);
            sendMessage.setMessage(message);
            queue.add(sendMessage);
            size.incrementAndGet();
        } catch (Exception e) {
            log.error("addToQueue: {}", e);
        }
    }

    /**
     * 统计topic上报数量
     * @param topic
     */
    public static void topicStatistics(String topic) {
        try {
            if (null == topicMap.get(topic)) {
                synchronized (topicMap) {
                    if (null == topicMap.get(topic)) {
                        topicMap.put(topic, new AtomicLong(0));
                    }
                }
            }
            increaseTopic(topic);
        } catch (Exception e) {
            log.error("topicStatistics: ", e);
        }
    }

    public static void increaseTopic(String topic) {
        AtomicLong count = topicMap.get(topic);
        if (null == count) {
            return;
        }
        long oldCount = count.get();
        while (!count.compareAndSet(oldCount, oldCount+1)) {
            oldCount = count.get();
        }
    }

    private void printTopicStatistics() {
        try {
            List<String> list = new ArrayList<>();
            for (String topic : topicMap.keySet()) {
                AtomicLong count = topicMap.get(topic);
                list.add(String.format("%s: %s", topic, count.get()));
            }
            log.info("MQTT发送消息统计: ", list.stream().collect(Collectors.joining(",")));
        } catch (Exception e) {
            log.error("printTopicStatistics: {}", e);
        }
    }




}
